package iterator;


import heap.*;
import global.*;
import bufmgr.*;
import diskmgr.*;
import index.*;

import java.lang.*;
import java.util.ArrayList;
import java.io.*;

import va.BaseRSIndexScan;
import va.BaseVAScan;
import va.Range;
import va.VAEntry;
import va_btree.RSBTIndexScan;
import va_btree.VABTIndexScan;
import va_btree.VABTreeFile;
import va_btree.VABTreeScan;
import va_file.VAFile;
import va_file.VAFileScan;
import va.BitSet;
/** 
 *
 *  This file contains an implementation of the nested loops join
 *  algorithm as described in the Shapiro paper.
 *  The algorithm is extremely simple:
 *
 *      foreach tuple r in R do
 *          foreach tuple s in S do
 *              if (ri == sj) then add (r, s) to the result.
 */

public class IndexNLJ  extends Iterator 
{
	private AttrType      _in1[],  _in2[];
	private   int        in1_len, in2_len;
	private   Iterator  outer;
	private   short t2_str_sizescopy[];
	private   CondExpr OutputFilter[];
	private   CondExpr RightFilter[];
	private   int        n_buf_pgs;        // # of buffer pages available.
	private   boolean        done,         // Is the join complete
	get_from_outer;                 // if TRUE, a tuple is got from outer
	private   Tuple     outer_tuple, inner_tuple;
	private   Tuple     Jtuple;           // Joined tuple
	private   FldSpec   perm_mat[];
	private   int        nOutFlds;
	private   Heapfile  hf;
	
	
	//Index properties -- used exclusively for index NLJ
	private VABTreeFile btreeFile;
	private VAFile vaFile;
	private VABTreeScan btreeScan;
	private RSBTIndexScan btScan;
	private VAFileScan vaScan;
	private IndexType indexType;
	private IDXSchema idxSchema;
	private BaseVAScan scan;
	private ArrayList<RID> rids;
	private Vector100Dtype target;
	private int distance;
	//private String relName, indexName;
	
	private   Scan      inner;
	private	Sort	innerSort;
	private boolean resetSort;

	/** constructor
	 * @author Mark Yates 
	 * New constructor to support Distance Join on vectors
	 *Initialize the two relations which are joined, including relation type,
	 *@param in1  Array containing field types of R.
	 *@param t1_str_sizes shows the length of the string fields.
	 *@param in2  Array containing field types of S
	 *@param  t2_str_sizes shows the length of the string fields.
	 *@param amt_of_mem  IN PAGES
	 *@param am1  access method for left i/p to join
	 *@param am2  access method (Sort) for right i/p to join
	 *@param outFilter   select expressions
	 *@param rightFilter reference to filter applied on right i/p
	 *@param proj_list shows what input fields go where in the output tuple
	 *@exception IOException some I/O fault
	 *@exception NestedLoopException exception from this class
	 * @throws InvalidTupleSizeException 
	 */	
	public IndexNLJ( AttrType    in1[],    
			short   t1_str_sizes[],
			AttrType    in2[],         
			short   t2_str_sizes[],   
			int     amt_of_mem,        
			Iterator     am1,          
			Sort 		 am2,      
			CondExpr outFilter[],      
			CondExpr rightFilter[],    
			FldSpec   proj_list[]
			) throws IOException, NestedLoopException, InvalidTupleSizeException
	{
		this(in1, in1.length, t1_str_sizes, in2, in2.length, t2_str_sizes, 
				amt_of_mem, am1, null, null, null, outFilter, rightFilter, proj_list, proj_list.length);

		innerSort = am2;
		resetSort = false;
	}

	/**constructor
	 *Initialize the two relations which are joined, including relation type,
	 *@param in1  Array containing field types of R.
	 *@param len_in1  # of columns in R.
	 *@param t1_str_sizes shows the length of the string fields.
	 *@param in2  Array containing field types of S
	 *@param len_in2  # of columns in S
	 *@param  t2_str_sizes shows the length of the string fields.
	 *@param amt_of_mem  IN PAGES
	 *@param am1  access method for left i/p to join
	 *@param relationName  access heapfile for right i/p to join
	 *@param outFilter   select expressions
	 *@param rightFilter reference to filter applied on right i/p
	 *@param proj_list shows what input fields go where in the output tuple
	 *@param n_out_flds number of outer relation fileds
	 *@exception IOException some I/O fault
	 *@exception NestedLoopException exception from this class
	 * @throws InvalidTupleSizeException 
	 */
	public IndexNLJ( AttrType    in1[],    
			int     len_in1,           
			short   t1_str_sizes[],
			AttrType    in2[],         
			int     len_in2,           
			short   t2_str_sizes[],   
			int     amt_of_mem,        
			Iterator     am1,          
			String relationName,
			IndexType indexType,
			String indexName,
			CondExpr outFilter[],      
			CondExpr rightFilter[],    
			FldSpec   proj_list[],
			int        n_out_flds
			) throws IOException,NestedLoopException, InvalidTupleSizeException
	{

		_in1 = new AttrType[in1.length];
		_in2 = new AttrType[in2.length];
		System.arraycopy(in1,0,_in1,0,in1.length);
		System.arraycopy(in2,0,_in2,0,in2.length);
		in1_len = len_in1;
		in2_len = len_in2;
		
		//Index-specific properties
		this.indexType = new IndexType(indexType.indexType);
		

		outer = am1;
		t2_str_sizescopy =  t2_str_sizes;
		inner_tuple = new Tuple();
		Jtuple = new Tuple();
		OutputFilter = outFilter;
		RightFilter  = rightFilter;

		n_buf_pgs    = amt_of_mem;
		inner = null;
		done  = false;
		get_from_outer = true;

		AttrType[] Jtypes = new AttrType[n_out_flds];
		short[]    t_size;
		
		//somehow get from OutputFilter -- does this work?
		distance = OutputFilter[0].distance; 

		perm_mat = proj_list;
		nOutFlds = n_out_flds;
		try {
			t_size = TupleUtils.setup_op_tuple(Jtuple, Jtypes,
					in1, len_in1, in2, len_in2,
					t1_str_sizes, t2_str_sizes,
					proj_list, nOutFlds);
		}catch (TupleUtilsException e){
			throw new NestedLoopException(e,"TupleUtilsException is caught by NestedLoopsJoins.java");
		}


		// Mark Yates - relationName may not be provided for distance join. A Sort will be used
		// instead of a heapfile scan. need to change to allow relationName to be null.
		if (relationName != null)
		{
			try {
				hf = new Heapfile(relationName);
	
			}
			catch(Exception e) {
				throw new NestedLoopException(e, "Cannot get schema and/or heap file.");
			}
			
		}
		

		try {
			switch(this.indexType.indexType)
			{
			case IndexType.VA_BT_Index:
				this.idxSchema = IDXSchema.GetIndex(indexName);
				btreeFile = new VABTreeFile(indexName, this.idxSchema.Bits);
				
			break;
			
			case IndexType.VA_Index:
				this.idxSchema = IDXSchema.GetIndex(indexName);
				vaFile = new VAFile(indexName);
				
			break;
			}
		}
		catch(Exception e) {
			throw new NestedLoopException(e, "Create index failed.");
		}
		
	
	}

	/**  
	 *@return The joined tuple is returned
	 *@exception IOException I/O errors
	 *@exception JoinsException some join exception
	 *@exception IndexException exception from super class
	 *@exception InvalidTupleSizeException invalid tuple size
	 *@exception InvalidTypeException tuple type not valid
	 *@exception PageNotReadException exception from lower layer
	 *@exception TupleUtilsException exception from using tuple utilities
	 *@exception PredEvalException exception from PredEval class
	 *@exception SortException sort exception
	 *@exception LowMemException memory error
	 *@exception UnknowAttrType attribute type unknown
	 *@exception UnknownKeyTypeException key type unknown
	 *@exception Exception other exceptions

	 */
	public Tuple get_next()
			throws IOException,
			JoinsException ,
			IndexException,
			InvalidTupleSizeException,
			InvalidTypeException, 
			PageNotReadException,
			TupleUtilsException, 
			PredEvalException,
			SortException,
			LowMemException,
			UnknowAttrType,
			UnknownKeyTypeException,
			Exception
			{
		// This is a DUMBEST form of a join, not making use of any key information...


		if (done)
			return null;

		do
		{
			// If get_from_outer is true, Get a tuple from the outer, delete
			// an existing scan on the file, and reopen a new scan on the file.
			// If a get_next on the outer returns DONE?, then the nested loops
			//join is done too.

			if (get_from_outer == true)
			{
				get_from_outer = false;
				// Mark Yates - moved logic to reset the inner loop's reader to resetInner()

				if ((outer_tuple=outer.get_next()) == null)
				{
					done = true;
					if(vaScan != null) {vaScan.close(); vaScan = null;}
					if(btreeFile != null) {btreeFile.close();}
					if(btreeScan != null) {btreeScan.close(); btreeScan = null;}
					
					else if(inner != null) {inner.closescan(); inner = null;}
					
					
					return null;
				}  
				
				//Does this get the right vector?
				int fldNo = OutputFilter[0].operand1.symbol.offset;
				target = outer_tuple.get100DVectFld(fldNo);
				resetInner();
				
			}  // ENDS: if (get_from_outer == TRUE)

                
			
			if(this.indexType.indexType == IndexType.None) //no index, so just use inner scan
			{
				while ((inner_tuple = getNextTupleFromInner()) != null){
					
					//use PredEval.Eval as Mark wrote to test output filter
					inner_tuple.setHdr((short)in2_len, _in2,t2_str_sizescopy);
					if (PredEval.Eval(RightFilter, inner_tuple, null, _in2, null) == true)
					{
						if (PredEval.Eval(OutputFilter, outer_tuple, inner_tuple, _in1, _in2) == true)
						{
							// Apply a projection on the outer and inner tuples.
							Projection.Join(outer_tuple, _in1, 
									inner_tuple, _in2, 
									Jtuple, perm_mat, nOutFlds);
							return Jtuple;
						}
					}
				}
			}
			
			else{
				//if(testTuple) //this is a candidate vector, so we must test the tuple for a join
	            while(!rids.isEmpty()) //iterate through the RIDs, and return on join condition
				{
	            	RID rid = rids.remove(0);
	            	
					//get inner tuple -- NOTE we may not want to use hf.getRecord if there are page-read issues
					//inner_tuple = hf.getRecord(entry.rid);
					inner_tuple = hf.getRecord(rid);
					
					//use PredEval.Eval as Mark wrote to test output filter
					inner_tuple.setHdr((short)in2_len, _in2,t2_str_sizescopy);
					if (PredEval.Eval(RightFilter, inner_tuple, null, _in2, null) == true)
					{
						if (PredEval.Eval(OutputFilter, outer_tuple, inner_tuple, _in1, _in2) == true)
						{
							// Apply a projection on the outer and inner tuples.
							Projection.Join(outer_tuple, _in1, 
									inner_tuple, _in2, 
									Jtuple, perm_mat, nOutFlds);
							return Jtuple;
						}
					}
				}
			}

		// There has been no match. (otherwise, we would have 
		//returned from t//he while loop. Hence, inner is 
		//exhausted, => set get_from_outer = TRUE, go to top of loop

			get_from_outer = true; // Loop back to top and get next outer tuple.	      
		} while (true);
			} 

	/**
	 * implement the abstract method close() from super class Iterator
	 *to finish cleaning up
	 *@exception IOException I/O error from lower layers
	 *@exception JoinsException join error from lower layers
	 *@exception IndexException index access error 
	 */
	public void close() throws JoinsException, IOException,IndexException 
	{
		if (!closeFlag) {

			try {
				outer.close();
			}catch (Exception e) {
				throw new JoinsException(e, "NestedLoopsJoin.java: error in closing iterator.");
			}
			closeFlag = true;
		}
	}
	/*
	 * Mark Yates
	 * Added this method to support using either a heapfile scan (original) or a
	 * Sort (new for distance join) for the inner loop.
	 * 
	 * This method resets the Scan or Sort to the beginning for the next iteration of the
	 * out loop.
	 */
	private void resetInner() 
			throws NestedLoopException
	{
		if ( hf != null )
		{
			try {
				switch(this.indexType.indexType) //if using an index, get the RIDs for candidates
				{
					case IndexType.VA_BT_Index:
						if(btreeScan != null) btreeScan.close();
						btreeScan = new VABTreeScan(btreeFile);
						rids = btreeScan.VABTreeRangeScanNoKeys(target, distance, idxSchema.Bits);
					break;
					
					case IndexType.VA_Index:
						scan = new VAFileScan(vaFile);
						rids = scan.VARangeScan(target, distance, idxSchema.Bits);
					break;
					
					case IndexType.None:
						inner = hf.openScan();
					break;}
			}
			catch(Exception e){
				throw new NestedLoopException(e, "openScan failed");
			}
		}
	}
	/*
	 * Mark Yates
	 * Added this method to support using either a heapfile scan (original) or a
	 * Sort (new for distance join) for the inner loop.
	 * 
	 * This method gets the next tuple for the inner loop.
	 */
	private Tuple getNextTupleFromInner() 
			throws SortException, UnknowAttrType, LowMemException, JoinsException, Exception
	{
		if (inner!=null)
		{
			// this is the original behavior
			RID rid = new RID();
			return inner.getNext(rid);
		}
		else
		{
			// this gets the next tuple from a Sort instead. used for the new Distance join
			return innerSort.get_next();
		}
	}
}
